package com.xbstar.datafusion.controller;

import java.util.List;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;
import com.xbstar.datafusion.databus.DataBusManage;
import com.xbstar.datafusion.store.StoreService;

@Component
public class DataController {
	@Autowired
	RuleManage ruleManage;
	@Autowired
	StoreService storeService;
	@Autowired
	DataBusManage dataBusManage;
	@KafkaListener(topics = "controllerget")
    public void processMessage(String content) {
		/**
		 * 此处是数据接受控制器，主要步骤有：
		 * 1. 加载相应规则
		 * 2. 按照规则进行拆包
		 * 3. 按照规则进行存储
		 * 4. 按照规则进行转发
		 */
        System.out.println(content);
        
        JSONObject json=JSONObject.parseObject(content);
        String sysId=json.getString("id");
        
        List<RuleDomain>rules=ruleManage.getRulesById(sysId);
        rules.parallelStream().forEach((ruleDomain)->{
        	if(ruleManage.check(json,ruleDomain)){
        		JSONObject jsonp= ruleManage.getDetailPackage(json,ruleDomain);
        		storeService.storeByRule(jsonp,ruleDomain);
        		List<JSONObject>senPs=ruleManage.getPackagesByRule(jsonp,ruleDomain);
        		senPs.parallelStream().forEach((pkg)->{
        			dataBusManage.send2System(pkg);
        		});
        		
        	}
        });
        
    }

}
